package com.study.chapter09;

import com.study.chapter05.source.ClickSource;
import com.study.entity.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/16 21:57
 */
public class TtlSetTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner((event, l) -> event.timestamp)
                );

        stream.keyBy(event -> event.user)
                .flatMap(new MyFlatMapResult())
                .print();

        env.execute();
    }

    //自定义flatMapFunction
    public static class MyFlatMapResult extends RichFlatMapFunction<Event,String> {
        //ValueState
        private ValueState<Event> valueState;

        @Override
        public void open(Configuration parameters) throws Exception {
            //定义状态描述器
            ValueStateDescriptor<Event> stateDescriptor = new ValueStateDescriptor<>("my-value-event", Event.class);
            //定义状态
            valueState = getRuntimeContext().getState(stateDescriptor);

            StateTtlConfig ttlConfig = StateTtlConfig
                    //设置ttl时间为1h
                    .newBuilder(Time.hours(1))
                    //只有创建或写操作才更新失效时间
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    //不返回过期值
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();
            //给状态描述器设置ttl
            stateDescriptor.enableTimeToLive(ttlConfig);
        }

        @Override
        public void flatMap(Event value, Collector<String> out) throws Exception {
            System.out.println(valueState.value());
            valueState.update(value);
            System.out.println("My Value："+valueState.value());
        }
    }
}
